package com.sunday.common.mq.rocket.brave.study.base.e11_PULL;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;

/**
 * 消费者数量 与 读队列数量 的关系
 * <p>
 * 读写队列作用
 * RocketMQ创建Topic时，可以配置writeQueueNums和readQueueNums，读写队列，是在做路由信息时使用。
 * 在消息发送时，使用写队列个数返回路由信息，而消息消费时按照读队列个数返回路由信息。在物理文件层面，只有写队列才会创建文件。
 * <p>
 * 写队列个数是8，设置的读队列个数是4，这个时候，会创建8个文件夹，代表0 1 2 3 4 5 6 7，但在消息消费时，路由信息只返回4，
 * 在具体拉取消息时，就只会消费0 1 2 3这4个队列中的消息，4 5 6 7中的信息压根就不会被消费。反过来，如果写队列个数是4，读队列个数是8，在生产消息时只会往0 1 2 3中生产消息，
 * 消费消息时则会从0 1 2 3 4 5 6 7所有的队列中消费，当然 4 5 6 7中压根就没有消息 。
 * <p>
 * readQueueNums>=writeQueueNums,程序才能正常进行
 * 最佳实践是readQueueNums=writeQueueNums。那rocketmq为什么要区分读写队列呢？直接强制readQueueNums=writeQueueNums，
 * 不就没有问题了吗？rocketmq设置读写队列数的目的在于方便队列的缩容和扩容。
 * <p>
 * 假如一个topic在每个broker上创建了128个队列，现在需要将队列缩容到64个，怎么做才能100%不会丢失消息，并且无需重启应用程序？
 * 最佳实践：先缩容写队列128->64，写队列由0 1 2 ......127缩至 0 1 2 ........63。等到64 65 66......127中的消息全部消费完后，
 * 再缩容读队列128->64(同时缩容写队列和读队列可能会导致部分消息未被消费)。
 */
@Slf4j
public class PullConsumer {

//    @PostConstruct
//    public void init() {
//        ClientConfig clientConfig = new ClientConfig();
//        MQClientInstance mqClientInstance = MQClientManager.getInstance().getOrCreateMQClientInstance(clientConfig);
//        log.info("{}", mqClientInstance);
//        mqClientInstance.getMQAdminImpl().createTopic();
//    }

    public static void main(String[] args) throws MQClientException {

//        consumerReadQueue();

//        readWriteQueue();

        multiConsumer();
    }

    /**
     * 测试情况 : 单消费者组，多消费者拉取测试 （写队列 = 读队列个数）
     * A. 消费者数量 5 > 读队列数量 4 -> 业务正常, 消费总数量没有受到影响，当中存在一个无用的消费者，符合rocket定义
     * B. 消费者数量 3 > 读队列数量 4 -> 业务正常, 存在某一个消费承载两条读队列的信息
     */
    public static void consumerReadQueue() throws MQClientException {
        for (int i = 0; i < 3; i++) {
            PullConsumerThread pullConsumerThread = new PullConsumerThread("127.0.0.1:9876", "pull_consumer_group_name", "PullTopic");
            pullConsumerThread.start();
        }
    }

    /**
     * 测试情况 : 读写队列不平等情况
     * <p>
     * 写队列个数是8，设置的读队列个数是4，这个时候，会创建8个文件夹，代表0 1 2 3 4 5 6 7，但在消息消费时，路由信息只返回4，
     * 在具体拉取消息时，就只会消费0 1 2 3这4个队列中的消息，4 5 6 7中的信息压根就不会被消费。反过来，如果写队列个数是4，读队列个数是8，在生产消息时只会往0 1 2 3中生产消息，
     * 消费消息时则会从0 1 2 3 4 5 6 7所有的队列中消费，当然 4 5 6 7中压根就没有消息 。
     * <p>
     * 经过测试 写 4 读 4
     * A。写 4  = 读 4 -> 全部队列使用率 100% ，无空闲
     * B。写 3 < 读 4  -> 有消息的队列信息 = 写的队列数量, 存在无用读队列, 直接影响了消费者的消费。
     * C。写 4 > 读 3  -> 存在部分消息丢失情况，写入无报错，消费者获取消息存在部分丢失
     * D。写 0 -> 发送端报错 如下:
     * Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No route info of this topic: PullTopic
     * E. 读 0 -> 消费端报错 如下:
     * Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: Can not find Message Queue for this topic, PullTopic
     * 但是任然可以启动, 但是生产端不会有任何一个消息可以递送
     */
    public static void readWriteQueue() throws MQClientException {
        for (int i = 0; i < 4; i++) {
            PullConsumerThread pullConsumerThread = new PullConsumerThread("127.0.0.1:9876", "pull_consumer_group_name", "PullTopic");
            pullConsumerThread.start();
        }
    }

    /**
     * 测试情况 : 多消费组，针对一个主题进行消费，是否存在重复消费情况
     * 多消费组会广播形式获取到全量的主题信息
     * PS: 在web页面进行操作删除 其中一个消费者组， 主题中任然存在
     */
    public static void multiConsumer() throws MQClientException {

        /**
         * 集群消费（Clustering）
         * 集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。
         *
         * 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。
         * 可以通过加机器，或者在已有机器启动多个进程的方式。
         *
         * 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。
         *
         * 关于消费者和订阅
         * 第一件需要注意的事情是，不同的消费者组可以独立的消费一些 topic，并且每个消费者组都有自己的消费偏移量，
         * 请确保同一组内的每个消费者订阅信息保持一致。
         */

        for (int i = 0; i < 2; i++) {
            PullConsumerThread pullConsumerThread = new PullConsumerThread("127.0.0.1:9876", "pull_consumer_group_name" + i, "PullTopic");
            pullConsumerThread.start();
        }
    }

}
